package com.lagou.rpc.consumer.client;

import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.zookeeper.handler.RpcRegistryCenter;
import com.lagou.rpc.zookeeper.handler.RpcRegistryCenterFactory;
import com.lagou.rpc.zookeeper.listener.NodeChangeListener;
import com.lagou.rpc.zookeeper.metrics.RequestMetrics;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 消费方
 *  用于获取所有服务提供端节点信息
 */
@Service
public class RpcConsumer implements NodeChangeListener {

    //用于每个存储rpcClient连接地址 key:com.lagou.edu.api.UserService/127.0.0.1:7777  value: 127.0.0.1:7777
    private static final Map<String, String> RPC_CLIENT_POOL = new ConcurrentHashMap<>();

    RpcRegistryCenter rpcRegistryCenter;

    public RpcConsumer() throws Exception {

        //获取需要消费的服务
        Map<String, Object> zkServiceMap = new ConcurrentHashMap<>();
        zkServiceMap.put(IUserService.class.getName(), IUserService.class);

        //获取注册中心
        if(null == rpcRegistryCenter){
            rpcRegistryCenter = new RpcRegistryCenterFactory().getObject();
        }

        zkServiceMap.entrySet().forEach(new Consumer<Map.Entry<String, Object>>() {
            @Override
            public void accept(Map.Entry<String, Object> entry) {
                String serviceName = entry.getKey();
                //发现com.lagou.rpc.api.IUserService服务  返回服务地址127.0.0.1:7777组合列表
                List<String> discovery = rpcRegistryCenter.discovery(serviceName);

                //遍历当前服务下需要连接的服务地址
                for (String connectString : discovery) {
                    RPC_CLIENT_POOL.put(serviceName+"/"+connectString,connectString);
                }
            }
        });
        rpcRegistryCenter.addListener(this);
    }

    /**
     * 用于处理节点变更信息
     * @param service     服务名
     * @param serviceList  当前服务节点下的子节点列表
     * @param pathChildrenCacheEvent
     */
    @Override
    public void notify(String service, List<String> serviceList, PathChildrenCacheEvent pathChildrenCacheEvent) {

        //节点变更事件类型
        PathChildrenCacheEvent.Type eventType = pathChildrenCacheEvent.getType();
        //变更节点数据
        ChildData data = pathChildrenCacheEvent.getData();
        if(null == data){
            return;
        }
        System.out.println("收到节点变更通知:事件类型" + eventType + "---服务" + service + "---子节点列表" + serviceList);
        //获取变更节点
        String path = data.getPath();
        String connectString = path.substring(path.lastIndexOf("/") + 1);
        String key = service+"/"+connectString;

        // 增加节点
        if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(eventType)) {

            //重新放入
            RPC_CLIENT_POOL.put(key,connectString);
            System.out.println("新增节点:" + key);
        }
        //移除节点
        if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(eventType)) {

            if(null != RPC_CLIENT_POOL && !RPC_CLIENT_POOL.isEmpty()){

                RPC_CLIENT_POOL.remove(key);
                System.out.println("移除节点:" + key);

                String[] split = connectString.split(":");
                RequestMetrics.getInstance().removeNode(split[0], Integer.parseInt(split[1]));
            }
        }

    }
    public Map<String, String> getRpcClientPool() {
        return RPC_CLIENT_POOL;
    }

}
